package event0208

import (
	"errors"
	"sync"
	"sync/atomic"
)

// 配置参数：worker 数量与 taskQueue 容量，根据场景调整
const (
	numWorkers    = 4
	taskQueueSize = 512
)

// Bus 表示事件总线，采用异步分发消息
type Bus struct {
	subNodes  sync.Map      // topic -> *node 映射
	taskQueue chan asyncTask // 异步任务队列
	closed    int32         // 0：未关闭，1：已关闭（原子标记）
	wg        sync.WaitGroup // 用于等待 worker 退出
}

// NewBus 初始化一个新的 Bus，并启动 worker 池
func NewBus() *Bus {
	bus := &Bus{
		taskQueue: make(chan asyncTask, taskQueueSize),
	}
	for i := 0; i < numWorkers; i++ {
		bus.wg.Add(1)
		go bus.worker()
	}
	return bus
}

// asyncTask 封装向某个订阅者发送消息的任务，采用值传递降低额外分配
type asyncTask struct {
	sub *Sub
	msg interface{}
}

// worker 从 taskQueue 中读取任务并调用订阅者的 receive 发送消息
func (b *Bus) worker() {
	defer b.wg.Done()
	for task := range b.taskQueue {
		task.sub.receive(task.msg)
	}
}

// Subscribe 为指定 topic 添加订阅者 sub
func (b *Bus) Subscribe(topic string, sub *Sub) {
	actual, _ := b.subNodes.LoadOrStore(topic, newNode())
	n := actual.(*node)
	n.addSub(sub)
}

// UnSubscribe 取消指定 topic 下订阅者 sub，若该 topic 下无订阅者则删除节点
func (b *Bus) UnSubscribe(topic string, sub *Sub) {
	actual, ok := b.subNodes.Load(topic)
	if !ok {
		return
	}
	n := actual.(*node)
	n.removeSub(sub)
	if n.subsLen() == 0 {
		b.subNodes.Delete(topic)
	}
}

// Publish 异步发布消息 msg 到指定 topic 下的所有订阅者
// 为避免 taskQueue 满而阻塞发布者，采用非阻塞写入：
// 如果 taskQueue 写入失败，则回退采用独立 goroutine 异步调用 sub.receive。
func (b *Bus) Publish(topic string, msg interface{}) error {
	if atomic.LoadInt32(&b.closed) == 1 {
		return errors.New("bus is closed")
	}
	actual, ok := b.subNodes.Load(topic)
	if !ok {
		return errors.New("topic not exist")
	}
	n := actual.(*node)
	subs := n.getSubs()
	for _, sub := range subs {
		select {
		case b.taskQueue <- asyncTask{sub: sub, msg: msg}:
			// 成功入队
		default:
			// taskQueue 满，启动单独 goroutine 发送消息，避免阻塞
			go sub.receive(msg)
		}
	}
	return nil
}

// PubFunc 返回便于闭包调用的发布函数
func (b *Bus) PubFunc(topic string) func(msg interface{}) {
	return func(msg interface{}) {
		_ = b.Publish(topic, msg)
	}
}

// SubsLen 返回指定 topic 的订阅者数量
func (b *Bus) SubsLen(topic string) (int, error) {
	actual, ok := b.subNodes.Load(topic)
	if !ok {
		return 0, errors.New("topic not exist")
	}
	n := actual.(*node)
	return n.subsLen(), nil
}

// Close 关闭 Bus，步骤如下：
// 1. 原子标记为关闭；
// 2. 关闭 taskQueue，等待所有 worker 退出；
// 3. 遍历所有 topic，关闭对应订阅者的接收通道。
func (b *Bus) Close() {
	if !atomic.CompareAndSwapInt32(&b.closed, 0, 1) {
		return
	}
	close(b.taskQueue)
	b.wg.Wait()
	b.subNodes.Range(func(key, value interface{}) bool {
		n := value.(*node)
		n.closeAll()
		return true
	})
}

////////////////////////////////////////////////////////////////////////
// node 及订阅者管理部分（采用 atomic.Pointer 降低类型断言开销）
////////////////////////////////////////////////////////////////////////

// node 表示一个主题节点，内部存储该 topic 下的所有订阅者列表
type node struct {
	mu   sync.Mutex
	subs atomic.Pointer[[]*Sub] // 持有 []*Sub 的指针
}

// newNode 初始化一个新的 node，并预分配小容量切片以减少后续扩容
func newNode() *node {
	n := &node{}
	initial := make([]*Sub, 0, 4)
	n.subs.Store(&initial)
	return n
}

// addSub 添加订阅者 sub，写路径上加锁，复制切片后更新
func (n *node) addSub(sub *Sub) {
	n.mu.Lock()
	defer n.mu.Unlock()
	current := *n.subs.Load()
	newSubs := append(current, sub)
	n.subs.Store(&newSubs)
}

// removeSub 移除订阅者 sub
func (n *node) removeSub(sub *Sub) {
	n.mu.Lock()
	defer n.mu.Unlock()
	current := *n.subs.Load()
	for i, s := range current {
		if s == sub {
			newSubs := append(current[:i], current[i+1:]...)
			n.subs.Store(&newSubs)
			return
		}
	}
}

// getSubs 读取当前订阅者切片（无锁）
func (n *node) getSubs() []*Sub {
	return *n.subs.Load()
}

// subsLen 返回当前订阅者数量
func (n *node) subsLen() int {
	return len(*n.subs.Load())
}

// closeAll 关闭该节点下所有订阅者的接收通道
func (n *node) closeAll() {
	for _, sub := range n.getSubs() {
		sub.close()
	}
}

////////////////////////////////////////////////////////////////////////
// Sub 订阅者部分
////////////////////////////////////////////////////////////////////////

// Sub 表示订阅者，内部持有用于接收消息的缓冲通道
type Sub struct {
	out chan interface{}
}

// NewSub 返回一个新的订阅者，默认缓冲大小为 1024
func NewSub() *Sub {
	return &Sub{
		out: make(chan interface{}, 1024),
	}
}

// receive 将消息 msg 送入订阅者通道，采用阻塞写入；
// 为避免因通道关闭引发 panic，这里通过 recover 忽略异常。
func (s *Sub) receive(msg interface{}) {
	defer func() { recover() }()
	s.out <- msg
}

// Out 返回订阅者的只读接收通道
func (s *Sub) Out() <-chan interface{} {
	return s.out
}

// close 关闭订阅者的接收通道
func (s *Sub) close() {
	close(s.out)
}
